package com.easy.mq.client;

import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.easy.mq.exception.MQException;
import com.easy.mq.result.KafkaSendResult;
import com.easy.mq.result.RabbitProducerMessage;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;

public class RabbitClient {

	private final Logger log = LoggerFactory.getLogger(RabbitClient.class);

	private Map<String, Object> properties;
	
	public static final String TOPIC_KEY = "default.topic";

	public static RabbitClient build() {
		return new RabbitClient();
	}

	public RabbitClient() {

	}

	public RabbitClient(Map<String, Object> properties) {
		this.properties = properties;
	}

	public KafkaSendResult send(RabbitProducerMessage message) throws MQException {
		
		try{
			if(StringUtils.isNotEmpty(TOPIC_KEY)) {
				properties.put(TOPIC_KEY, TOPIC_KEY);
			}
			Channel channel = RabbitProducerPool.getProducer(properties);
			channel.queueDeclare(message.getQueue(),true,false,false,null);
			channel.basicPublish("",message.getQueue(),
	                    MessageProperties.PERSISTENT_TEXT_PLAIN,message.getContent().getBytes());
			return null;
		} catch (Exception e) {
			log.error(e.getMessage(), e);
			RabbitProducerPool.removePool(properties);
			throw new MQException(e.getMessage());
		} 

	}

}
